package kr.or.javacafe.twitter;

import java.io.IOException;
import java.util.Map;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.type.TypeReference;

public class Fun {

	
	/*
	 * MapReduce Mapper
	 *
	 * 주어진 텍스트를 단어 단위로 쪼갠다음 프레임워크로 보낸다.
	 * 
	 * Key : 단어
	 * Value : 1
	 */
	public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
		
		// Key 
		private Text word = new Text();
		// Value
		private final static LongWritable one = new LongWritable(1);
		
		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String strJson = value.toString();

			try {
				ObjectMapper om = new ObjectMapper();
				Map<String, Object> objTempMap = om.readValue(strJson, new TypeReference<Map<String, Object>>() {});
				
				String strText = (String)objTempMap.get("text");				
				
				// 트윗문장을 단어로 분리하여 처리한다.
				StringTokenizer token = new StringTokenizer(strText);
				while (token.hasMoreElements()) {
					word.set(token.nextToken().toLowerCase());
					
					context.write(word, one);
				}
				
			} catch (Exception ex) {
				ex.printStackTrace();
			}				
		}
	}
	
	
	/*
	 * MapReduce Reducer
	 * 
	 * 단어들의 카운트를 sum 처리하여 집계한다.
	 */
	public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
		
		private LongWritable sumWritable = new LongWritable();
		
		public void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
			
			// 단어의 리스트를 sum 처리한다.
			long sum = 0;
			for (LongWritable val : values) {
				sum += val.get();
			}
			sumWritable.set(sum);
			
			context.write(key, sumWritable);
		}
	}
	
	
	/*
	 * Main
	 * 
	 * RapReduce 설정을 구성하여 프로그램을 실행한다.
	 */
	public static void main(String[] args) throws Exception {
		
		Configuration conf = new Configuration();
		
		// Job Alias
		Job job = new Job(conf, "WordCount");
		
		// 로드 할 함수명 정의
		job.setJarByClass(Fun.class);
		job.setMapperClass(MyMapper.class);
		job.setReducerClass(MyReducer.class);
		
		// 파라메터 타입 정의
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.class);
		
		// Raw 데이터 타입 정의
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		// Raw 데이터 정보
		FileInputFormat.addInputPath(job, new Path(args[0]));
		// Result 데이터 정보
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		// 처리시간동안 blocking (stdout true)
		job.waitForCompletion(true);
	}
	
}




